/*
**    Copyright (C) 2003-2011 Institute for Systems Biology
**                            Seattle, Washington, USA.
**
**    This library is free software; you can redistribute it and/or
**    modify it under the terms of the GNU Lesser General Public
**    License as published by the Free Software Foundation; either
**    version 3 of the License, or (at your option) any later version.
**
**    This library is distributed in the hope that it will be useful,
**    but WITHOUT ANY WARRANTY; without even the implied warranty of
**    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
**    Lesser General Public License for more details.
**
**    You should have received a copy of the GNU Lesser General Public
**    License along with this library; if not, write to the Free Software
**    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*/

package org.systemsbiology.rface.hadoop.mappers;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
//import java.util.logging.Logger;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.systemsbiology.hadoop.ioformats.LexicalKey;
import org.systemsbiology.rface.hadoop.ConfigConstants;
import org.systemsbiology.rface.hadoop.data.TargetRecord;

import org.systemsbiology.util.Filter;
import org.systemsbiology.util.ConjunctionFilter;
import org.systemsbiology.util.Pair;

public class LineToTargetRecord extends Mapper<Text, Text, LexicalKey, TargetRecord>{
	private int[] valueIndexes = null;
	private int targetIndex = -1;
	private int predictorIndex = -1;
	private HashMap<Integer, HashSet<String>> kills = null;
	
	protected ConjunctionFilter<Pair<String, String>> featureFilter =null;
	
	protected void setup(Context context){
		Configuration config = context.getConfiguration();
		String[] valueColumns = config.get(ConfigConstants.VALUE_COLUMNS).split("\\s");
		valueIndexes = new int[valueColumns.length];
		for(int i = 0; i < valueIndexes.length; ++i){
			valueIndexes[i] = Integer.parseInt(valueColumns[i]);
		}
		
		String predictorColumn = config.get(ConfigConstants.PREDICTOR_COLUMN, "1");
		predictorIndex = Integer.parseInt(predictorColumn);
		String targetColumn = config.get(ConfigConstants.TARGET_COLUMN, "0");
		targetIndex = Integer.parseInt(targetColumn);
		
		kills = new HashMap<Integer, HashSet<String>>();
		String[] killCmds = config.get(ConfigConstants.KILL_SEGMENTS, "").split("\\s");
		for(String killme: killCmds){
			String[] thing = killme.split(Pattern.quote(":"));
			int col = Integer.parseInt(thing[0]);
			if(!kills.containsKey(col)){
				kills.put(col, new HashSet<String>());
			}
			kills.get(col).add(thing[1]);
			//Logger.getLogger(LineToTargetRecord.class.getName()).info("Killing " + thing[1] + " on group " + thing[0]);
		}
		featureFilter = new ConjunctionFilter<Pair<String, String>>();
		featureFilter.add(new UnDead());
		//Logger.getLogger(LineToTargetRecord.class.getName()).info("Primary setup complete");
	}
	
	protected class UnDead implements Filter<Pair<String, String>>{
		public boolean accept(Pair<String, String> p){
			//Logger log = Logger.getLogger(LineToTargetRecord.class.getName());
			//log.info("Filtering for anti-deadness of " + p.toString());
			if(dead(p.first)){
				//log.info("==> First was DEAD.");
				return false;
			}
			if(dead(p.second)){
				//log.info("==> Second was DEAD.");
				return false;
			}
			//log.info("==> Both features alive.");
			return true;
		}
	}
	
	protected boolean dead(String feature){
		//Logger log = Logger.getLogger(LineToTargetRecord.class.getName());
		//log.info("Calculating deadness of " + feature);
		String[] chunks = feature.split(Pattern.quote(":"));
		for(int k = 0; k < chunks.length; ++k){
			HashSet<String> condemnations = kills.get(k);
			if(condemnations == null) continue;
			if(condemnations.contains(chunks[k])){
				//log.info("==> DEAD.");
				return true;
			}
		}
		//log.info("==> Alive.");
		return false;
	}
	
	protected void map(Text key, Text value, Context target) throws InterruptedException, IOException{
		String[] columns=value.toString().split("\\t");
		if(!featureFilter.accept(new Pair<String, String>(columns[targetIndex], columns[predictorIndex]))) return;
		for(int vDex : valueIndexes){
			TargetRecord rec = new TargetRecord();
			boolean negative = false;
			if(vDex < 0){
				vDex = -vDex;
				negative = true;
				if(vDex == Integer.MIN_VALUE){
					vDex = 0; //special case for negative zero
				}
			}
			double colValue = Double.parseDouble(columns[vDex]);
			if(negative) colValue = -colValue;
			rec.addPredictor(key.toString(), columns[predictorIndex], colValue);
			
			target.write(new LexicalKey(new String[]{columns[targetIndex], String.valueOf(vDex)}), rec);
		}
	}
}
